package com.cctc.rds.scheduler.server.schedule.hashedwheel;

import java.io.Closeable;
import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 时间轮。
 */
public class HashedWheelTimer implements Runnable, Closeable {
    private static final int INITIAL = 0, STARTED = 1, SHUTDOWN = 2;
    private final AtomicInteger status = new AtomicInteger(INITIAL);
    private volatile long startTime = 0;
    private final CountDownLatch hasStarted = new CountDownLatch(1);
    private final LinkedBlockingQueue<HashedWheelBucket.HashedWheelTimeout> queue = new LinkedBlockingQueue<>();
    private final int n = 60;
    private final HashedWheelBucket[] wheel = new HashedWheelBucket[n];
    private TimerTaskCallback timerTaskCallback;
    private Thread timer;

    public HashedWheelTimer(TimerTaskCallback timerTaskCallback) {
        this.timerTaskCallback = timerTaskCallback;
        this.timer = new Thread(this, "hashed-wheel-timer");
        this.timer.setDaemon(true);
    }

    public synchronized void addTask(String taskKey, long scheduleTime, long execTime) {
        start();
        TimerTask task = new TimerTask(taskKey);
        long deadline = (scheduleTime + execTime) - startTime;
        if (deadline > 0) {
            HashedWheelBucket.HashedWheelTimeout timeout = new HashedWheelBucket.HashedWheelTimeout(task, deadline);
            queue.offer(timeout);
            return;
        }
        timerTaskCallback.exec(Collections.singletonList(task));
    }

    private void start() {
        switch (status.get()) {
            case INITIAL:
                if (status.compareAndSet(INITIAL, STARTED)) {
                    timer.start();
                }
            case STARTED:
                break;
            case SHUTDOWN:
                throw new IllegalStateException("时间轮已关闭！！！");
            default:
                throw new IllegalStateException("无效状态！！！");
        }
        while (startTime == 0) {
            try {
                hasStarted.wait();
            } catch (InterruptedException e) {
            }
        }
    }

    @Override
    public void run() {
        startTime = System.currentTimeMillis() / 1000;
        hasStarted.countDown();
        while (status.get() != SHUTDOWN) {
            processBlockingQueue();
            execTask();
        }
    }

    private void processBlockingQueue() {
        int count = 0;
        while (count < 100000) {
            try {
                HashedWheelBucket.HashedWheelTimeout timeout = queue.poll(10, TimeUnit.MILLISECONDS);
                if (timeout == null) return;
                count++;
                long deadline = timeout.getDeadline();
                int rounds = (int) (deadline / n);
                timeout.setRounds(rounds);
                int index = (int) (deadline % n);
                if (wheel[index] == null) wheel[index] = new HashedWheelBucket();
                wheel[index].addTimeout(timeout);
            } catch (InterruptedException e) {
            }
        }
    }

    private void execTask() {
        long curDeadline = System.currentTimeMillis() / 1000 - startTime;
        int curRounds = (int) (curDeadline / n);
        int index = (int) (curDeadline % n);
        if (wheel[index] != null) {
            List<TimerTask> tasks = wheel[index].removeTimeout(curRounds);
            if (tasks.isEmpty()) return;
            timerTaskCallback.exec(tasks);
        }
    }

    @Override
    public void close() throws IOException {
        status.set(SHUTDOWN);
    }

}
